package com.erbao.io.practice;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * selector 的 线程
 * @author CaoBaoYin
 * @since 2022/10/10 上午8:12
 */
public class SelectorThread implements Runnable {
    /*每个线程对应一个selector。
    * 多线程情况下，并发的客户端分配到多个selector上
    * 每个客户端 只绑定到其中一个selector上 之间没有交互*/

    Selector selector = null;

    // 线程间通信
    LinkedBlockingQueue<Channel> lbq = new LinkedBlockingQueue<>();

    // 因为 acceptHandler 需要注册客户端，而当前selector thread并不知道其他线程的存在，需要借助SelectorThreadGroup
    SelectorThreadGroup stg = null;

    public SelectorThread(SelectorThreadGroup stg) {
        try {
            this.stg = stg;
            selector = Selector.open();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void run() {
        // loop 一直转圈做事
        while (true) {
            try {
                // 1. select
                // select() 阻塞 ，其他线程如果注册到了selector上，这个线程并不能立即感知到
                // 所以假设select调用时，没有注册事件就会发生阻塞，即使其他线程注册了事件当前线程也无法感知到。这样就会一直阻塞，此时需要方法wakeup()唤醒
                System.out.println(Thread.currentThread().getName()+"：selector.select前……"+selector.keys().size());
                int num = selector.select();
                System.out.println(Thread.currentThread().getName()+"：selector.select后……"+selector.keys().size());
                if (num > 0) {
                    //有事件
                    // 2. selectkeys
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> it = keys.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();// 这里进行移除，因为set集合每次使用的是同一个，不移除下次调用selectedKeys依然会有这个事件
                        if (key.isAcceptable()) {// 接收客户端之后，需要分配到不同的selector上。相对复杂
                            acceptHandler(key);
                        } else if (key.isReadable()) {
                            readHandler(key);
                        } else if (key.isWritable()){

                        }
                    }
                }

                // 3.处理一些task
                if (! this.lbq.isEmpty()) {
                    Channel ch = lbq.take();// 栈是独立的，堆是共享的。方法的逻辑、本地变量是隔离的。
                    if (ch instanceof ServerSocketChannel) {
                        ServerSocketChannel server = (ServerSocketChannel) ch;
                        server.register(selector, SelectionKey.OP_ACCEPT);
                        System.out.println(Thread.currentThread().getName()+": register listen");
                    } else if (ch instanceof SocketChannel) {
                        SocketChannel client = (SocketChannel) ch;

                        ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
                        client.register(selector, SelectionKey.OP_READ, buffer);

                        System.out.println(Thread.currentThread().getName()+": register client: "+client.getRemoteAddress());
                    }
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private void readHandler(SelectionKey key) throws IOException {
        // 自我认知：为什么要使用客户端附件形式携带buffer，而不是每次重新开辟？
        // 1.每次开辟需要时间，性能损耗；
        // 2.并不一定每次都能开辟成功，一旦开辟失败，会造成无法处理这个客户端。应用服务处理能力变差
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        SocketChannel client = (SocketChannel) key.channel();
        buffer.clear();
        while (true) {
            /*简单处理，发什么返回什么*/
            // 读取内容到buffer
            int length = client.read(buffer);
            if (length > 0) {
                buffer.flip();// 内容反转，直接写出
                while (buffer.hasRemaining()) {
                    client.write(buffer);// 读写操作时，pos和limit都会变动，所以此处没有常规写法的index
                }
                buffer.clear();//使用前清理，使用后清理，因为buffer是在重复利用，为了保证buffer没有其他的内容，只有程序想要的内容
            } else if (length == 0) {
                break;// 如果能读到内容就一直读，直到读完为止
            } else if (length < 0) {
                // 可能客户端断开了
                System.out.println(Thread.currentThread().getName()+": client: "+ client.getRemoteAddress() +"closed……");
                key.cancel();
                client.close();// 老师单线程代码有关闭
                break;
            }
        }
    }

    private void acceptHandler(SelectionKey key) {
        System.out.println(Thread.currentThread().getName()+": acceptHandler……");
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        try {
            SocketChannel client = server.accept();
            client.configureBlocking(false);

            // 如果但线程就直接进行注册了
            // choose a selector and register.
            stg.nextSelectorV3(client);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
